package com.jjzhk.common.beanConfig;

import com.jjzhk.common.PuddoService;
import com.jjzhk.common.bean.RpcRequest;
import com.jjzhk.common.bean.RpcResponse;
import com.jjzhk.common.handler.RpcDecoder;
import com.jjzhk.common.handler.RpcEncoder;
import com.jjzhk.common.handler.RpcServerHandler;
import com.jjzhk.common.utils.ProtoStuffSerializationUtils;
import com.jjzhk.common.utils.ReflexUtils;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.config.BeanFactoryPostProcessor;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.beans.factory.support.BeanDefinitionRegistry;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.context.annotation.ClassPathBeanDefinitionScanner;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.core.type.filter.AnnotationTypeFilter;

import java.net.InetAddress;
import java.util.HashMap;
import java.util.Map;

/**
 * ServiceBean
 *
 * @author : JJZHK
 * @date : 2016-08-15
 * @comments :
 **/
public class ServiceBean implements BeanFactoryPostProcessor,
        DisposableBean, ApplicationContextAware,BeanPostProcessor, ApplicationListener<ApplicationEvent> {
    private Logger logger = LoggerFactory.getLogger(ServiceBean.class);
    private Map<String, Object> handlerMap = new HashMap<>();
    private String annoPackage;
    private String SerializationClass;

    private String[] packages;
    private int port;

    private RegistryBean zookeeperBean;

    public String getSerializationClass() {
        return SerializationClass;
    }

    public void setSerializationClass(String serializationClass) {
        SerializationClass = serializationClass;
    }

    public int getPort() {
        return port;
    }

    public void setPort(int port) {
        this.port = port;
    }

    public String getAnnoPackage() {
        return annoPackage;
    }

    public void setAnnoPackage(String annoPackage) {
        this.annoPackage = annoPackage;
        if(!StringUtils.isEmpty(annoPackage))
            this.packages = annoPackage.split(",");
    }

    private ApplicationContext ctx;

    @Override
    public void setApplicationContext(ApplicationContext ctx) throws BeansException {
        this.ctx = ctx;
    }

    @Override
    public Object postProcessAfterInitialization(Object bean, String beanName)
            throws BeansException {
        if (bean instanceof ApplicationBean)
        {
            this.port = ((ApplicationBean) bean).getPort();
        }
        if (bean instanceof RegistryBean)
        {
            zookeeperBean = (RegistryBean) bean;
        }
        if(!this.isMatchPackage(bean))return bean;

        PuddoService rpcService = bean.getClass().getAnnotation(PuddoService.class);
        if(rpcService != null){
            String serviceName = rpcService.value().getName(); // interface name
            String serviceVersion = rpcService.uuid();
            serviceName += "-" + serviceVersion;
            handlerMap.put(serviceName, bean); // serviceBean is the interface implement class
            logger.info("handlerMap is : {}, {}", serviceName, bean);
        }

        return bean;
    }

    @Override
    public Object postProcessBeforeInitialization(Object bean, String beanName)
            throws BeansException {
        return bean;
    }

    @Override
    public void destroy() throws Exception {

    }

    @Override
    public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException {
        if (packages == null || packages.length == 0) {
            return;
        }

        if (beanFactory instanceof BeanDefinitionRegistry) {
            BeanDefinitionRegistry beanDefinitionRegistry = (BeanDefinitionRegistry)beanFactory;
            ClassPathBeanDefinitionScanner scanner = new ClassPathBeanDefinitionScanner(beanDefinitionRegistry,true);

            AnnotationTypeFilter filter = new AnnotationTypeFilter(PuddoService.class);
            scanner.addIncludeFilter(filter);

            scanner.scan(packages);
        }
    }

    private boolean isMatchPackage(Object bean){
        if (packages == null || packages.length == 0) {
            return true;
        }
        String beanClassName = bean.getClass().getName();
        for (String pkg : packages) {
            if (beanClassName.startsWith(pkg)) {
                return true;
            }
        }
        return false;
    }

    @Override
    public void onApplicationEvent(ApplicationEvent event) {
        if (event instanceof ContextRefreshedEvent) {

            EventLoopGroup bossGroup = new NioEventLoopGroup();
            EventLoopGroup workGroup = new NioEventLoopGroup();

            try {
                ServerBootstrap serverBootstrap = new ServerBootstrap();
                serverBootstrap.group(bossGroup, workGroup).channel(NioServerSocketChannel.class).option(ChannelOption.SO_BACKLOG, 1024);
                serverBootstrap.childOption(ChannelOption.SO_KEEPALIVE, true);
                serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel channel) throws Exception {
                        Class<?> forSerializationClass = null;
                        if (StringUtils.isEmpty(SerializationClass))
                        {
                            forSerializationClass = ProtoStuffSerializationUtils.class;
                        }
                        else
                        {
                            forSerializationClass = ReflexUtils.forName(SerializationClass);
                        }
                        channel.pipeline().addLast("decoder", new RpcDecoder(RpcRequest.class, forSerializationClass)); // deconder, inbound
                        channel.pipeline().addLast("encoder", new RpcEncoder(RpcResponse.class, forSerializationClass)); // encoder, outbound
                        channel.pipeline().addLast("handler", new RpcServerHandler(handlerMap)); // handler, inbound
                    }
                });

                ChannelFuture future = serverBootstrap.bind(port).sync();
                logger.info("server starts on port : {}", port);

                // 向zookeeper注册
                InetAddress addr = InetAddress.getLocalHost();
                String ip=addr.getHostAddress().toString();//获得本机IP
                zookeeperBean.registryService(handlerMap, String.format("%s:%d", ip, port));

                future.channel().closeFuture().sync();
            }
            catch (Exception ex)
            {
                logger.error("{}", ex);
            }
            finally {
                workGroup.shutdownGracefully();
                bossGroup.shutdownGracefully();
            }
        }
    }
}
